package com.aotain.project.apollo.boltv2;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.commons.codec.binary.Base64;
import org.apache.commons.math3.distribution.NormalDistribution;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseDao;
import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.DosSliding;
import com.aotain.apollo.SlidingWindow;
import com.aotain.apollo.utils.Constants;
import com.aotain.storm.AbstractBolt;
import com.aotain.storm.TupleHelpers;

public class UDIIAbnormalCalBoltV2 extends AbstractBolt {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = 1L;

	private int windowLengthInSeconds;
	private int emitFrequencyInSeconds;
	private HBaseDao dao = null;

	private HashMap<String,Double> dipPInfo = new HashMap<String,Double>();//各个目标Ip的熵值
	private HashMap<String,Long> dipPVMap = new HashMap<String,Long>();//目标IP PV数
	private HashMap<String,Long> sdPVMap = new HashMap<String,Long>();//各个源ip到目标ip的PV数
	private Map<String,Long> tripleMap = new HashMap<String,Long>();//各个元组的PV
	private Map<String,Set<String>> dipToTriple = new HashMap<String,Set<String>>();//目标ip下的三元组映射
	//	private Set<String> IPSet = new HashSet<String>();//要过滤的ip
	private Map<String,Integer> abnTimeMap = new HashMap<String,Integer>();//持续异常时间数map
	private Set<String> abnDipSet = new HashSet<String>();//信息熵异常的Dip
	private Map<String,Double> confidValueMap = new HashMap<String,Double>();//异常判断的最大置信区间
	private Map<String,Long> delaytimeMap = new HashMap<String,Long>();//求延时Map
	private Map<String,Long> maxDelayMap = new HashMap<String,Long>();//最大延时Map
	private Map<String,Long> minDelayMap = new HashMap<String,Long>();//最小延时Map
	private Map<String,ArrayList<Double>> dipHurstMap = new HashMap<String,ArrayList<Double>>();//存放每个dip所有时间窗口的熵值 
	private Map<String,Integer> streamPacketMap = new HashMap<String,Integer>();//
	private Map<String,Integer> delayTimePVMap = new HashMap<String,Integer>();//

	private String dip = null;
	private String dport = null;
	private String sip = null;
	private String sport = null;
	private byte[] gisbytes =  null;
	private String gisstr = null;
	private String tripleKey = null;
	private String sdipKey = null;  
	private String sourceAreaId = null;
	private String sourceAreaCountry = null;
	private String destAreaName = null;
	private String destGis = null;
	private String sourceAreaName = null;
	private String sourceGis = null;
	private String areaName = null;
	private String sourceAreaProvinceId = null;
	private String sourceAreaCityId = null;
	private String type = null;//协议类型
	private String idcno = null;
	private String idc_dip = null;
	private Integer upStreamOctets = 0;
	private Integer upStreamPacket = 0;
	private long delaytime = 0;
	private String avgdelay = null;
	private long sdMaxdelay = 0;
	private long sdMindelay = 0;
	private SlidingWindow cache;
	private DosSliding sdpvCache;
	private DosSliding dipPvCache;

	public UDIIAbnormalCalBoltV2() {
		this(Constants.WindowLengthInSeconds,Constants.EmitFrequencyInSeconds);
	}

	//	public WindowCalBolt(HashMap<Long,IPDatabase> map,ApolloConfig ap) {
	//		this.map = map;
	//		this.ap = ap;
	//	}   

	public UDIIAbnormalCalBoltV2(int windowLengthInSeconds,int emitFrequencyInSeconds ){
		if(windowLengthInSeconds%emitFrequencyInSeconds!=0){
			System.err.println("窗口个数要是整数");
		}
		this.windowLengthInSeconds = windowLengthInSeconds;
		this.emitFrequencyInSeconds = emitFrequencyInSeconds;
		//		cache = new SlidingWindow(windowLengthInSeconds/emitFrequencyInSeconds);
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60);
		return conf;
	}


	//计算需要的值
	private void countInLocal(Tuple tuple) {

		dip = tuple.getString(0);
		dport = tuple.getString(1);
		sip = tuple.getString(2);
		sport = tuple.getString(3);
		type = tuple.getString(7);
		delaytime = Long.parseLong(tuple.getString(8));
		idcno = tuple.getString(9);
		upStreamOctets = tuple.getInteger(10);
		upStreamPacket = tuple.getInteger(11);
		gisbytes =  Base64.decodeBase64(tuple.getString(4));
		gisstr = new String(gisbytes);
		tripleKey = dip + "*" + dport + "*" + sip + "*" + gisstr + "*" + type + "*" + idcno;
		sdipKey = dip + "*" + idcno + "*" + sip;   
		idc_dip = dip + "*" + idcno;

		//		if(IPSet.contains(dip)) { 
		//		if("80".equals(dport)) {
		if(Integer.parseInt(dport)<10000) {
			long dipPV = upStreamOctets;
			long triplePV = upStreamOctets;
			long sdPV = upStreamOctets;
			int triplePacket = upStreamPacket;
			int delayPV = 1;
			long sumtime = 0;
			//			long maxdelay = 0;
			//			long mindelay = 0;

			if(dipPVMap.get(idc_dip) != null) {
				dipPV = dipPVMap.get(idc_dip) + upStreamOctets;
			}
			if(tripleMap.get(tripleKey) != null) {
				triplePV = tripleMap.get(tripleKey) + upStreamOctets;
			}
			if(sdPVMap.get(sdipKey) != null) {
				sdPV = sdPVMap.get(sdipKey) + upStreamOctets;
			}
			if(streamPacketMap.get(tripleKey) != null) {
				triplePacket = streamPacketMap.get(tripleKey) + upStreamPacket;
			}
			if(delaytimeMap.get(sdipKey) != null) {
				sumtime = delaytimeMap.get(sdipKey) + delaytime;
			}else {
				sumtime = delaytime ;
			}
			if(delayTimePVMap.get(sdipKey) != null) {
				delayPV = delayTimePVMap.get(sdipKey) + 1;
			}
			if(maxDelayMap.get(sdipKey) == null) {
				maxDelayMap.put(sdipKey, delaytime);
			}
			if(delaytime>maxDelayMap.get(sdipKey)) {
				maxDelayMap.put(sdipKey, delaytime);
			}
			if(minDelayMap.get(sdipKey) == null) {
				minDelayMap.put(sdipKey, delaytime);
			}
			if(delaytime<minDelayMap.get(sdipKey)) {
				minDelayMap.put(sdipKey, delaytime);
			}

			Set<String> tripleSet = dipToTriple.get(idc_dip);
			if(tripleSet == null) {
				tripleSet = new HashSet<String>();
			}
			tripleSet.add(tripleKey);
			dipToTriple.put(idc_dip, tripleSet);

			sdPVMap.put(sdipKey, sdPV);
			dipPVMap.put(idc_dip, dipPV);
			tripleMap.put(tripleKey, triplePV);
			delaytimeMap.put(sdipKey, sumtime);
			streamPacketMap.put(tripleKey, triplePacket);
			delayTimePVMap.put(sdipKey, delayPV);
		}
	}

	//定时发送
	private void emitCountingData(OutputCollector collector) {

		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);

		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmmss");
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_m1  = new SimpleDateFormat("yyyyMMddHHmm00");
		SimpleDateFormat sdf_m2  = new SimpleDateFormat("yyyyMMddHHmm");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");  
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
		sdpvCache.addPInfo(sdPVMap);
		dipPvCache.addPInfo(dipPVMap);
		//求熵
		for(Entry<String,Set<String>> entry:dipToTriple.entrySet()) {
			String dip_idc = entry.getKey();
			Set<String> tripleSet = entry.getValue();
			Long dipPV = dipPVMap.get(dip_idc);
			double pInfo = 0.0;
			for(Iterator<String> iter=tripleSet.iterator();iter.hasNext();) {
				String triple = iter.next();
				double p = (double)tripleMap.get(triple)/(double)dipPV;
				double tmpInfo = -Math.log(p)*p;
				pInfo += tmpInfo;
			}
			dipPInfo.put(dip_idc, pInfo);
		}

		cache.addPInfo(dipPInfo);
		Map<Integer,HashMap<String,Double>> infoMap = cache.getTuplesWindow();//得到滑动窗口中关于熵的数据
		Map<String,Integer> pInfoLengthMap = new HashMap<String,Integer>();//求熵值平均值时的长度
		Map<String,Double> totalP = new HashMap<String,Double>();//当前所有窗口对应目标元组的熵值和的Map
		Map<String,Double> avgP = new HashMap<String,Double>();//目标元组熵值的均值
		if(infoMap != null ) {
			for(HashMap<String,Double> singleMap: infoMap.values()) {
				if(singleMap != null) {
					for(Entry<String,Double> entry : singleMap.entrySet()) {
						String dip_idc = entry.getKey();
						Double value = entry.getValue();
						Double totalPvalue = totalP.get(dip_idc);
						//						if(value != 0) {
						Integer sum = pInfoLengthMap.get(dip_idc);
						if(sum == null) {
							sum = 0;
						}
						sum ++;
						pInfoLengthMap.put(dip_idc, sum);
						if(totalPvalue == null) {
							totalPvalue = value;
						} else {
							totalPvalue += value;
						}
						totalP.put(dip_idc, totalPvalue);
					}
				}
			}
		}

		for(String dkey : totalP.keySet()) {
			double avgPvalue = totalP.get(dkey) / (double) pInfoLengthMap.get(dkey);
			avgP.put(dkey, avgPvalue);
		}

		HashMap<String,Double> subsqrMap = new HashMap<String,Double>();
		for(String key : avgP.keySet()) {
			double value = 0.0;
			if(dipPInfo.get(key) != null && avgP.get(key)!= null) {
				value = (dipPInfo.get(key)-avgP.get(key))*(dipPInfo.get(key)-avgP.get(key));
			}
			subsqrMap.put(key, value);
		}
		cache.addSubsqr(subsqrMap);

		Map<String,Double> totalSubsqr = new HashMap<String,Double>();//差的平方和汇总
		Map<String,Double> stdMap = new HashMap<String,Double>();//标准差集合
		Map<Integer,HashMap<String,Double>> winSums = cache.getSumWindow();
		Map<Integer,HashMap<String,Long>> abnNumMaps = sdpvCache.getTuplesWindow();


		for(HashMap<String,Double> singleMap: winSums.values()) {
			if(singleMap != null) {
				for(Entry<String,Double> entry : singleMap.entrySet()) {
					String key = entry.getKey();
					Double value = entry.getValue();
					Double totalSvalue = totalSubsqr.get(key);
					if(totalSvalue == null) {
						totalSvalue = 0.0;
					} else {
						totalSvalue += value;
					}
					totalSubsqr.put(key, totalSvalue);
				}
			}
		}

		double stdValue = 0.0;
		double sxn = 0.0;
		double confidenceValueMax = 0.0;//最大置信区间
		double confidenceValueMin = 0.0;//最小置信区间
		double nordis = 0.0;
		Double avg = 0.0;
		Double dipP = 0.0;
		NormalDistribution s = new NormalDistribution (0,1);//求正态分布分位数所用
		//公式是：S_n=√(1/(n-1) ∑_(i=1)^n?〖(x_i-x)〗^2 ) 、S_xn=S_n/√n
		//该循环跑两次，主要为了实现连续五分钟异常判断，第一次先得到当前窗口类所有异常dip，以便后面过滤不是连续异常的dip
		for(String key : totalSubsqr.keySet()) {
			if(pInfoLengthMap.get(key) != null && pInfoLengthMap.get(key) >1) {

				stdValue = Math.sqrt(totalSubsqr.get(key)/(pInfoLengthMap.get(key)-1));//S_n
				sxn = stdValue / Math.sqrt(pInfoLengthMap.get(key));//S_xn=S_n/√n
				nordis = s.cumulativeProbability(sxn);
				avg = avgP.get(key)==null?0.0:avgP.get(key);
				dipP = dipPInfo.get(key)==null?0.0:dipPInfo.get(key);
				confidenceValueMax = avg + nordis;

				if(dipP>confidenceValueMax && sxn !=0) {
					abnDipSet.add(key);
				}
				else
				{//如果信息熵未达标，将内存中保存的DIP清掉。
					abnTimeMap.remove(key);
				}
			} 
		}

		Set<String> tmpSet = new HashSet<String>();//用来排除不是连续三分钟出现异常的dip 所用的临时set
		for(Iterator<String> iter = abnTimeMap.keySet().iterator();iter.hasNext();) {
			tmpSet.add(iter.next());
		}		
		//		if(abnTimeMap.size()>0) {
		//			tmpSet = abnTimeMap.keySet();//这样下面remove会导致引用删掉
		//		}
		tmpSet.removeAll(abnDipSet);//得到差集
		for(Iterator<String> iter = tmpSet.iterator();iter.hasNext();) {
			String strv = iter.next();
			abnTimeMap.remove(strv);
			confidValueMap.remove(strv);//移除最大置信区间记录值
		}
		abnDipSet.clear();

		//		int n = 1;
		Set<String> hPvSet = new HashSet<String>();
		//公式是：S_n=√(1/(n-1) ∑_(i=1)^n?〖(x_i-x)〗^2 ) 、S_xn=S_n/√n
		for(String key : totalSubsqr.keySet()) {
			dip = key.split("\\*",-1)[0];
			idcno = key.split("\\*",-1)[1];
			if(pInfoLengthMap.get(key) != null && pInfoLengthMap.get(key) >1) {

				Date date = new Date();
				String dateStr_h = sdf_h.format(date);
				String dateStr_d = sdf_d.format(date);
				String dateStr_hour = sdf_hour.format(date);

				stdValue = Math.sqrt(totalSubsqr.get(key)/(pInfoLengthMap.get(key)-1));//S_n
				sxn = stdValue / Math.sqrt(pInfoLengthMap.get(key));//S_xn=S_n/√n
				nordis = s.cumulativeProbability(sxn);
				avg = avgP.get(key)==null?0.0:avgP.get(key);
				dipP = dipPInfo.get(key)==null?0.0:dipPInfo.get(key);
				confidenceValueMax = avg + nordis; 	
				confidenceValueMin = avg - nordis;


				Set<String> tripleSet = dipToTriple.get(key);
				String rowKeyAbnormal = "";//dip date sip
				String rowKeyNormal = "";//dip date sip
				String rowKeyAbnlog3 = "";//sip date

			
				//dateStr sip sport dip dport attribution PV
				if(tripleSet != null) {

					Integer abnTime = 0;
					boolean isNormal = true;
					//$$$$Modified by turk 2016-01-04 修正异常连续判断问题
					if(dipP>confidenceValueMax && sxn !=0) {
						//							abnDipSet.add(key);
						abnTime = abnTimeMap.get(key);
						if(abnTime == null) {
							abnTime = 1;
						} else {
							abnTime += 1;
						}
						abnTimeMap.put(key, abnTime);
						if(abnTime >= 3) { //满足大于3分钟 才算异常 并入库
							//验证是否高于最大置信区间
							isNormal = false;
							Logger.getLogger(UDIIAbnormalCalBoltV2.class).info("UDII Abnormal Is Normal fase" + key);
							if(confidValueMap.get(key) == null) {
								confidValueMap.put(key, confidenceValueMax);
								//								confidValueMap.put(key, confidenceValueMin);
							} else {
								double tmpMaxconfidValue = confidValueMap.get(key);
								if(tmpMaxconfidValue>confidenceValueMin) {//如果此时最小置信区间小于记录的最大置信区间 则判断为正常
									isNormal = true;
									confidValueMap.remove(key);
									abnTimeMap.remove(key);
								}
							}

							//***判断是不是连续3分钟pv数超过60
							if(!isNormal) {
								for(HashMap<String,Long> sdpvmap : abnNumMaps.values()) {
									Set<String> norSet = new HashSet<String>();
									for(Entry<String,Long> entry : sdpvmap.entrySet()) {
										String dsip = entry.getKey();
										long v = entry.getValue();
										if(v<600) {
											norSet.add(dsip);
										}
									}
									for(Iterator<String> iter=norSet.iterator();iter.hasNext();) {
										sdpvmap.remove(iter.next());
									}
								}
							}
							if(abnNumMaps.size()>2) {
								Map<String ,Long> map1 = abnNumMaps.get(0);
								Map<String ,Long> map2 = abnNumMaps.get(1);
								Map<String ,Long> map3 = abnNumMaps.get(2);
								for(Iterator<String> iter=map1.keySet().iterator();iter.hasNext();) {
									String dsip = iter.next();
									if(map2.containsKey(dsip) && map3.containsKey(dsip)) {
										hPvSet.add(dsip);
									}
								}
							}
						}
					}
					//$$$$Modified by turk 2016-01-04 ----END

					for(Iterator<String> iter = tripleSet.iterator();iter.hasNext();) {

						String triple = iter.next();
						String[] splits = triple.split("\\*",-1);
						String dport = splits[1];
						String gis = splits[3];
						String protoType = splits[4];
						String sip = splits[2];
						String sdIP = key + "*" + sip;

						String[] gisArray = gis.split("#",-1);
						destAreaName = gisArray[0];
						destGis = gisArray[1];
						sourceAreaName = gisArray[2];//市
						sourceGis = gisArray[3];
						sourceAreaCountry = gisArray[4];
						sourceAreaId = gisArray[5];
						areaName = gisArray[6];//省
						sourceAreaCityId = gisArray[7];
						sourceAreaProvinceId = gisArray[8];
						Date date_ = new Date();
						String dateStr = sdf.format(date_);
						String evaluate = "";
						String desc = "";
						String abnormal = "0";
						String delayTime = "";
						String rowkey_ = "";
						long attnum = 0;
						avgdelay = String.format("%d", delaytimeMap.get(sdIP)/delayTimePVMap.get(sdIP));


						if(!isNormal) { //满足大于三分钟 才算异常 并入库
							//							isNormal = false;

							if(hPvSet.contains(sdIP)) {
								rowKeyAbnlog3 = dateStr + "_" + sip;
								rowKeyAbnormal = dip + "_" + dateStr + "_" + sip + "_" + dport + "_" + 1;	
								rowkey_ = dip + "_" + dateStr + "_" + sip + "_" + dport + "_" + 1;	
								abnormal = "1";
								delayTime = "平均延迟："+ avgdelay + "s;最大延时：" + maxDelayMap.get(sdIP) + "s;最小延时：" + minDelayMap.get(sdIP)+"s;";
								attnum = sdPVMap.get(sdIP);

								boolean isAbnormal = true;
								if(attnum<=50) {
									evaluate = "100";
									desc = "连接";
									abnormal = "0";
									isAbnormal = false;
									isNormal = true;
								} else if(attnum>50 && attnum<=200) {
									abnormal = "1";
									evaluate = "60";
									desc = "异常连接,持续时间>---" + abnTimeMap.get(key) + "分钟";

								} else if(attnum>200) {
									evaluate = "40";
									abnormal = "1";
									desc = "异常连接,持续时间>---" + abnTimeMap.get(key) + "分钟";
								}
						
								if(isAbnormal) {//异常连接单独另入一张表
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEIP", sip);
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTPORT", dport);
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ACCESSTIME", dateStr);
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTIP", dip);
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEAREA", sourceAreaName);
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEGEO", sourceGis);
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCECOUNTRY", sourceAreaCountry);
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTAREA", destAreaName);
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTGEO", destGis);
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ATTNUM", attnum+"");
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ABRNORMAL", abnormal);
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESC", desc);
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "EVALUATE", evaluate);
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "PROVINCE", areaName);
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(tripleMap.get(triple)));	
									hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(streamPacketMap.get(triple)));


									//									System.err.println("............SDS_ABNORMAL_LOG3...........rk=" + rowKeyAbnlog3 + "AREAID" + sourceAreaProvinceId +",AREANAME=" + areaName + ",CITYID=" + sourceAreaCityId + ",CITYNAME" + sourceAreaName + ",TYPE=" +protoType);
									hbaseInstance.Add("SDS_ABNORMAL_LOG3", rowKeyAbnlog3, "cf", "ACCESSTIME", dateStr);
									hbaseInstance.Add("SDS_ABNORMAL_LOG3", rowKeyAbnlog3, "cf", "SOURCEIP", sip);
									hbaseInstance.Add("SDS_ABNORMAL_LOG3", rowKeyAbnlog3, "cf", "SOURCEGEO", sourceGis);
									hbaseInstance.Add("SDS_ABNORMAL_LOG3", rowKeyAbnlog3, "cf", "DESTGEO", destGis);
									hbaseInstance.Add("SDS_ABNORMAL_LOG3", rowKeyAbnlog3, "cf", "ATTNUM", attnum+"");
									hbaseInstance.Add("SDS_ABNORMAL_LOG3", rowKeyAbnlog3, "cf", "AREAID", sourceAreaProvinceId);
									hbaseInstance.Add("SDS_ABNORMAL_LOG3", rowKeyAbnlog3, "cf", "AREANAME", areaName);
									hbaseInstance.Add("SDS_ABNORMAL_LOG3", rowKeyAbnlog3, "cf", "CITYID", sourceAreaCityId);
									hbaseInstance.Add("SDS_ABNORMAL_LOG3", rowKeyAbnlog3, "cf", "CITYNAME", sourceAreaName);
									hbaseInstance.Add("SDS_ABNORMAL_LOG3", rowKeyAbnlog3, "cf", "DESTIP", dip);
									hbaseInstance.Add("SDS_ABNORMAL_LOG3", rowKeyAbnlog3, "cf", "DESTAREA", destAreaName);
									hbaseInstance.Add("SDS_ABNORMAL_LOG3", rowKeyAbnlog3, "cf", "DESTPORT", dport);
									hbaseInstance.Add("SDS_ABNORMAL_LOG3", rowKeyAbnlog3, "cf", "TYPE", protoType);
									hbaseInstance.Add("SDS_ABNORMAL_LOG3", rowKeyAbnlog3, "cf", "SOURCECOUNTRY", sourceAreaCountry);
								}

								collector.emit(new Values(dip,sip,dport,sport,dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,areaName,
										tripleMap.get(triple),dateStr_hour,"SESSIONMIDDLE",idcno,sourceAreaCountry,Long.valueOf(tripleMap.get(triple)),Long.valueOf(streamPacketMap.get(triple))));
								//								collector.emit("AbnormalCalBoltStream",new Values(key,sip,dport,sport,dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,areaName,tripleMap.get(triple),dateStr_hour,"SESSIONMIDDLE"));
							} else {
								isNormal = true;
							}
						} else { 
							isNormal = true;
						}
						if(isNormal) {
							//正常日志入库SDS_IDC_LOG     	
							rowkey_ = dip + "_" + dateStr + "_" + sip + "_" + dport + "_" + 0;	
							evaluate = "100";
							abnormal = "0";
							desc = "连接";
							attnum = sdPVMap.get(sdIP);
							delayTime = "平均延迟："+ avgdelay + "s;最大延时：" + maxDelayMap.get(sdIP) + "s;最小延时：" + minDelayMap.get(sdIP)+"s;";
							collector.emit("ABNORMAL_LOG", new Values(rowkey_,dip,sip,dport,dateStr,sourceAreaName,sourceGis,sourceAreaCountry,destAreaName,destGis,evaluate,abnormal,desc,attnum+"",delayTime ,"0"));
							Logger.getLogger(UDIIAbnormalCalBoltV2.class).info("Import Normal Log " + rowkey_);
						}

					}
				}
				//				n++;
			} 
			stdMap.put(key, stdValue);
		}
		tripleMap.clear();//各元组pv清零
		dipPVMap.clear();//dipPV清零
		dipPInfo.clear();
		dipToTriple.clear();
		sdPVMap.clear();
		delaytimeMap.clear();
		maxDelayMap.clear();
		minDelayMap.clear();
		delayTimePVMap.clear();
		dipHurstMap.clear();
		streamPacketMap.clear();
		cache.advanceHead();
		sdpvCache.advanceHead();
		dipPvCache.advanceHead();
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		//		declarer.declareStream("AbnormalCalBoltStream",new Fields("dip","sip","dport","sport","dateStr_d","dateStr_h","sourceAreaId","sourceAreaName","areaName","pv","dateStr_hour","abntype"));
		declarer.declare(new Fields("dip","sip","dport","sport","dateStr_d","dateStr_h",
				"sourceAreaId","sourceAreaName","areaName","pv","dateStr_hour","abntype","idcno","sourcecountry","streamoctets","streampacket"));
		declarer.declareStream("ABNORMAL_LOG", new Fields("rowkey","dip","sip","dport","accesstime","sourcearea",
				"sourcegeo","sourcecountry","destarea","destgeo","evaluate","abnormal","desc","attnum","delaytime","flag"));
	}

	@Override
	public void cleanup() {

	}


	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			if(TupleHelpers.isTickTuple(tuple)) {
				emitCountingData(collector);  	
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			System.err.println(" _windowcalbolt exception"+e.getMessage());
			Logger.getRootLogger().error("windowcalbolt exception====",e);
			collector.fail(tuple);
			throw new FailedException(e);
		}	
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		cache = new SlidingWindow(20);
		sdpvCache = new DosSliding(3);//源到目标 pv数滑动窗口（精确到异常源ip时所用）
		dipPvCache = new DosSliding(20);//目标ip pv数
	}
}
